package com.crazyice.lee.jobs;

import com.crazyice.lee.data.Student;
import com.crazyice.lee.reader.JdbcReader;
import com.crazyice.lee.writer.JdbcWriter;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

@Slf4j
public class Mysql2MysqlJob {

    public static void main(String[] args) throws Exception {
        //设置环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //获取数据-数据源
        DataStreamSource<Student> students = env.addSource(new JdbcReader());
        students.name("从mysql读取数据");

        //处理数据过程
        DataStream<Student> subStudents = students
                .filter(e -> e.isSex()).name("过滤男生")
                .filter(e -> e.getAge() > 12).name("过滤12岁以上");

        //写入数据-处理结果
        subStudents.addSink(new JdbcWriter()).name("写入备份学生库");
        env.execute("读取Mysql，数据过滤，写入Mysql");
    }
}
